package com.leo.tool.jdk.jdk8;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
public class CompletableFutureTest {

    private AtomicInteger count;

    private ExecutorService executor = Executors.newFixedThreadPool(4);

    @Before
    public void setup() {
        count = new AtomicInteger(0);
    }

    @After
    public void tearDown() {
        Thread.yield();
    }

    /* ******************** construct CompletableFuture ******************** */

    @Test
    public void runnable() {
        CompletableFuture.runAsync(() -> printCurrentThread());
        CompletableFuture.runAsync(() -> printCurrentThread(), executor);
    }

    @Test
    public void callable() {
        CompletableFuture.supplyAsync(() -> Thread.currentThread().getName()).thenAccept(System.out::println);
        CompletableFuture.supplyAsync(() -> Thread.currentThread().getName(), executor).thenAccept(System.out::println);
    }

    @Test
    public void completedFuture() {
        //1、使用runAsync或supplyAsync发起异步调用
        CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> "result1");
        //2、CompletableFuture.completedFuture()直接创建一个已完成状态的CompletableFuture
        CompletableFuture<String> cf2 = CompletableFuture.completedFuture("result2");
        //3、先初始化一个未完成的CompletableFuture，然后通过complete()、completeExceptionally()，完成该CompletableFuture
        CompletableFuture<String> cf = new CompletableFuture<>();
        cf.complete("success");
    }

    @Test
    public void all() {
        CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> printProcess());
        CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> printProcess(), executor);
        CompletableFuture.allOf(future1, future2).thenAccept(r -> Assert.assertEquals(2, count.intValue()));
    }

    @Test
    public void any() {
        CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> printProcess(), executor);
        CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> printProcess(), executor);
        CompletableFuture.anyOf(future1, future2).thenAccept(r -> Assert.assertEquals(1, count.intValue()));
    }

    /* ******************** difference when run whit async ******************** */

    @Test
    public void thenRun() {
        // 由于UniRun未传入executor，thenRun()方法不可能新启线程
        //
        // 短路处理（前一个任务已完成），717行执行uniRun()方法（任务由当前线程执行）
        // push到前一个CompletableFuture中，之后在postComplete方法执行tryFire()方法（任务由CompletableFuture所在线程执行）
        // 短路处理（前一个任务已完成），720行执行tryFire()方法（任务由当前线程执行）
        CompletableFuture.runAsync(() -> printCurrentThread()).thenRun(() -> printCurrentThread())
                .thenRun(() -> printCurrentThread());
        CompletableFuture.runAsync(() -> printCurrentThread(), executor)
                .thenRun(() -> printCurrentThread()).thenRun(() -> printCurrentThread());
    }

    // Async使用指定线程池（默认内部asyncPool）执行之后的任务
    @Test
    public void thenRunAsync() {
        // UniRun传入了指定executor(默认内部asyncPool)，thenRunAsync使用executor执行
        //
        // 短路处理（前一个任务已完成），720行执行tryFire()方法（任务由executor执行）
        // push到前一个CompletableFuture中，之后在postComplete方法执行tryFire()方法（任务由executor执行）
        CompletableFuture.runAsync(() -> printCurrentThread()).thenRunAsync(() -> printCurrentThread())
                .thenRunAsync(() -> printCurrentThread()); // ForkJoinPool可能会使用同一个线程？
        CompletableFuture.runAsync(() -> printCurrentThread(), executor)
                .thenRunAsync(() -> printCurrentThread(), executor).thenRunAsync(() -> printCurrentThread(), executor);
    }

    @Test
    @SneakyThrows
    public void afterGet() {
        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> printCurrentThread());
        future.get();
        future.thenRun(() -> printCurrentThread()); // main
        future.thenRunAsync(() -> printCurrentThread()); // ForkJoinPool
    }

    /* ******************** exception handle ******************** */

    // CompletableFuture是一条执行链，UniRun等包装了任务执行，异常通过completeThrowable封装到RESULT中，不会抛出异常
    // CompletableFuture链执行下一个节点任务后，RESULT通过completeNull清除（AltResult(null)）
    @Test
    public void exceptional() {
        // exception
        CompletableFuture.supplyAsync(() -> 1 / 0).exceptionally(exception -> {
            Assert.assertTrue(exception instanceof ArithmeticException);
            return 0;
        }).thenAccept(r -> Assert.assertEquals(0, r.intValue()));
        // normal(not run)
        CompletableFuture.supplyAsync(() -> 1 / 1).exceptionally(exception -> {
            Assert.assertTrue(exception instanceof ArithmeticException);
            return 0;
        }).thenAccept(r -> Assert.assertEquals(1, r.intValue()));
    }

    @Test
    public void handle() {
        // exception
        CompletableFuture.supplyAsync(() -> 1 / 0).handle((result, exception) -> {
            Assert.assertNull(result);
            Assert.assertTrue(exception instanceof ArithmeticException);
            return 0;
        }).thenAccept(r -> Assert.assertEquals(0, r.intValue()));
        // normal(run)
        CompletableFuture.supplyAsync(() -> 1 / 1).handle((result, exception) -> {
            Assert.assertNull(exception);
            return result;
        }).thenAccept(r -> Assert.assertEquals(1, r.intValue()));
    }

    @Test
    public void whenComplete() {
        // exception
        CompletableFuture.supplyAsync(() -> 1 / 0).whenComplete((result, exception) -> {
            Assert.assertNull(result);
            Assert.assertTrue(exception instanceof ArithmeticException);
        }).thenAccept(r -> Assert.assertNull(r));
        // normal(run)
        CompletableFuture.supplyAsync(() -> 1 / 1).whenComplete((result, exception) -> {
            Assert.assertEquals(1, result.intValue());
            Assert.assertNull(exception);
        }).thenAccept(r -> Assert.assertEquals(1, r.intValue()));
    }

    /* ******************** chains programming ******************** */

    // 函数式编程思路，在方法里new新对象，传入this为新对象dependency，新对象绑定到自身完成任务后回调
    @Test
    public void chains() {
        // thenAccept
        CompletableFuture.completedFuture("hello").thenApply(s -> "world")
                .thenAccept(r -> Assert.assertEquals("world", r));
        // thenRun
        setup();
        CompletableFuture.runAsync(() -> printProcess()).thenRun(() -> Assert.assertEquals(1, count));
        // thenCombine
        CompletableFuture.completedFuture("hello")
                .thenCombine(CompletableFuture.completedFuture("world"), (a, b) -> a + " " + b)
                .thenAccept(r -> Assert.assertEquals("hello world", r));
        // thenAcceptBoth
        CompletableFuture.completedFuture("hello").thenAcceptBoth(CompletableFuture.completedFuture("world"),
                (a, b) -> Assert.assertEquals("hello world", a + " " + b));
        // runAfterBoth
        setup();
        CompletableFuture.runAsync(() -> printProcess()).runAfterBoth(CompletableFuture.runAsync(() -> printProcess()),
                () -> Assert.assertEquals(2, count));
        // applyToEither(和anyof类似)
        CompletableFuture.completedFuture("hello").applyToEither(CompletableFuture.completedFuture("world"), a -> a)
                .thenAccept(r -> Assert.assertEquals("hello", r));
        // acceptEither
        CompletableFuture.completedFuture("hello").acceptEither(CompletableFuture.completedFuture("world"),
                r -> Assert.assertEquals("hello", r));
        // runAfterEither
        setup();
        CompletableFuture.runAsync(() -> printProcess())
                .runAfterEither(CompletableFuture.runAsync(() -> printProcess()), () -> Assert.assertEquals(1, count));
        // thenCompose
        CompletableFuture.completedFuture("hello")
                .thenCombine(CompletableFuture.completedFuture("world"), (a, b) -> a + " " + b)
                .thenAccept(r -> Assert.assertEquals("hello world", r));
        // exceptionally
        exceptional();
        // whenComplete
        whenComplete();
        // handle
        handle();
    }

    /* ******************** private methods ******************** */

    private void printCurrentThread() {
        System.out.println("current thread:" + Thread.currentThread().getName());
    }

    @SneakyThrows
    private int printProcess() {
        Thread.sleep(10);
        return count.incrementAndGet();
    }

}
